package etcd

import (
	"context"
	"fmt"
	"src/zeus/logger"
	"src/zeus/threads"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"go.etcd.io/etcd/api/v3/mvccpb"
	clientv3 "go.etcd.io/etcd/client/v3"
)

type option func(md *EtcdAccess)

// 设置ETCD地址
func SetEtcdAddrs(li ...string) option {
	return func(md *EtcdAccess) {
		md.config.Endpoints = li
	}
}

// 设置超时时间
func SetEtcdDialTimeout(t time.Duration) option {
	return func(md *EtcdAccess) {
		md.config.DialTimeout = t
	}
}

// 租约时间（秒）
func SetEtcdLeaseTTL(ttl int64) option {
	return func(md *EtcdAccess) {
		md.leaseTTL = ttl
	}
}

// 设置，当发生自己注册信息丢失的回调
func SetEtcdTimeoutFunc(f func()) option {
	return func(md *EtcdAccess) {
		md.timeoutFunc = f
	}
}

/*
etcd的连接器
*/
type EtcdAccess struct {
	etcd    *clientv3.Client  //etcd的连接
	config  clientv3.Config   //配置信息
	thgo    *threads.ThreadGo //监听管理器
	leaseID atomic.Int64      //本服务所有数据的租约
	// leaseID  clientv3.LeaseID  //本服务所有数据的租约
	leaseTTL int64    //租约时间（秒）
	reFuncs  []reFunc //如果发生租约过期了，需要重新注册的逻辑
	//发生自己注册信息丢失的回调
	timeoutFunc func()
	status      sync.WaitGroup //是否已经启动的状态
}

func NewEtcdAccess(opts ...option) (result *EtcdAccess, err error) {
	result = new(EtcdAccess)
	result.reFuncs = make([]reFunc, 0, 2)
	result.thgo = threads.NewThreadGo()
	result.config = clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	}
	result.leaseTTL = 50
	result.status.Add(1)
	for _, opt := range opts {
		opt(result)
	}
	//试试会不会返回错误
	if result.etcd, err = clientv3.New(result.config); err != nil {
		return nil, err
	}
	if err = result.watchLease(); err != nil {
		result.Close()
		return nil, err
	}
	return result, err
}

func (this *EtcdAccess) Start() {
	this.status.Done()
}

func (this *EtcdAccess) Close() {
	// this.status.Done()
	tmp := this.leaseID.Load()
	this.leaseID.Store(0)
	this.etcd.Lease.Revoke(this.thgo.Ctx, clientv3.LeaseID(tmp))
	this.etcd.Close()
	this.thgo.CloseWait()
}

// 监听某个目录，f方法为收到信息后的回调逻辑,回调新协程上
func (this *EtcdAccess) WatchPrefix(pathname string, f func(md *clientv3.Event)) {
	this.thgo.Go(func(ctx context.Context) {
		this.status.Wait()
		chli := this.etcd.Watch(ctx, pathname, clientv3.WithPrefix())
		for {
			select {
			case <-ctx.Done():
				return
			case etli := <-chli:
				logger.Debug(logger.LogKey_Etcd2, "etcd发现", pathname, "事件数量", len(etli.Events))
				for i, md := range etli.Events {
					f(md)
					logger.Debug(logger.LogKey_Etcd2, "    event", i, md.Type, md.Kv, md.PrevKv)
				}
			}
		}
	})
	return
}

// 获取指定路径下的所有数据
func (this *EtcdAccess) GetPathAll(pathname string) (*clientv3.GetResponse, error) {
	return this.etcd.Get(this.thgo.Ctx, pathname, clientv3.WithPrefix())
}

// 删除数据
func (this *EtcdAccess) Delete(key string) error {
	if _, err := this.etcd.Delete(this.thgo.Ctx, key); err != nil {
		return err
	}
	return nil
}

// 写入数据
func (this *EtcdAccess) Put(key, val string) error {
	if strings.HasSuffix(key, "/load") {
		logger.Debug(logger.LogKey_Etcd2, "本服务器上报etcd 负载信息:", key, val)
	} else {
		logger.Debug(logger.LogKey_Etcd, "本服务器上报etcd 服务信息:", key, val)
	}
	if _, err := this.etcd.Put(this.thgo.Ctx, key, val, clientv3.WithLease(clientv3.LeaseID(this.leaseID.Load()))); err != nil {
		return err
	}
	return nil
}

// 写入数据，永远
func (this *EtcdAccess) PutForever(key, val string) error {
	logger.Debug(logger.LogKey_Etcd, "PutForever 本服务器上报etcd信息:", key, val)
	if _, err := this.etcd.Put(this.thgo.Ctx, key, val); err != nil {
		return err
	}
	return nil
}

// 撤回指定Lease的Keep
func (this *EtcdAccess) RevokeLease(leaseid clientv3.LeaseID) {
	this.etcd.Lease.Revoke(this.thgo.Ctx, leaseid)
}

// 获取一个KEY的数据
func (this *EtcdAccess) Get(key string) (result *mvccpb.KeyValue, err error) {
	res, err := this.etcd.Get(this.thgo.Ctx, key)
	if err != nil {
		return nil, err
	}
	if len(res.Kvs) > 0 {
		return res.Kvs[0], nil
	}
	return nil, fmt.Errorf("Get data len == 0")
}

// 设置注册方法
func (this *EtcdAccess) SetReFunc(fs ...reFunc) {
	this.reFuncs = append(this.reFuncs, fs...)
}

// 生成LeaseID 并监视租约
// 租约过期后，会重新生成LeaseID并重新注册 etcd 信息
func (this *EtcdAccess) setLeaseID() (leasech <-chan *clientv3.LeaseKeepAliveResponse, err error) {
	lease, err := this.etcd.Lease.Grant(this.thgo.Ctx, this.leaseTTL)
	if err != nil {
		return nil, err
	}
	this.leaseID.Store(int64(lease.ID))
	// this.leaseID = lease.ID
	if leasech, err = this.etcd.Lease.KeepAlive(this.thgo.Ctx, lease.ID); err != nil {
		return nil, err
	} else {
		return leasech, nil
	}
}

// 监视Lease的有效性
// 如果租约过期，则重新生成LeaseID并重新注册 etcd 信息
func (this *EtcdAccess) watchLease() error {
	var leasech <-chan *clientv3.LeaseKeepAliveResponse
	var err error
	leasech, err = this.setLeaseID()
	if err != nil {
		return err
	}
	this.thgo.Go(func(ctx context.Context) {
		this.status.Wait()
		for {
			select {
			case _, ok := <-leasech: //keepAlive 每间隔租约时间，都会收到一个信号
				//有租赁的时候，发现失效了才做的逻辑
				if !ok && this.leaseID.Load() != 0 {

					// this.leaseID.Store(0)
					// this.RevokeLease(this.leaseID)
					/*
						因为调试等其他原因导致租约出现问题;
						需要重新开起租约，并发起原来的注册;
						如果注册失败，需要支持重试；
					*/

					for leasech, err = this.setLeaseID(); err != nil; leasech, err = this.setLeaseID() {
						logger.Errorf("setLeasedID. %v", err)
						time.Sleep(time.Second)
						select {
						case <-ctx.Done():
							return
						default:
						}
					}
					if this.timeoutFunc != nil {
						this.timeoutFunc()
						logger.Debug(logger.LogKey_Etcd, "etcd租赁 leaseID timeout. call func info.", this.timeoutFunc)
					} else {
						logger.Warn("Etcd leaseID timeout. no func info.")
					}
					//新租赁拿到后重新获取所有etcd的注册信息
					//如果这个过程中又断掉了，也会操作完了，再重新走逻辑
					for i := range this.reFuncs {
						this.reFuncs[i]()
						logger.Debug(logger.LogKey_Etcd, "etcd租赁 重新连接到etcd 注册自己的信息", i, this.reFuncs[i])
					}
					logger.Info("Etcd 重新监听！")
				}
			case <-ctx.Done():
				return
			}
		}
	})
	return nil
}
